package com.excu.factory.connection.heat;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class InterfaceServer {
    private static final int PORT = 9000;
    private static final int BUFFER_SIZE = 1024;
    private static final long HEARTBEAT_TIMEOUT = 60_000; // 60秒无心跳则断开

    // 服务器接口常量（与客户端对应）
    public static final String INTERFACE_USER = "/api/user";
    public static final String INTERFACE_ORDER = "/api/order";
    public static final String INTERFACE_PRODUCT = "/api/product";

    // NIO组件
    private final Selector selector;
    private final ServerSocketChannel serverChannel;
    private final ByteBuffer readBuffer = ByteBuffer.allocate(BUFFER_SIZE);
    private final ByteBuffer writeBuffer = ByteBuffer.allocate(BUFFER_SIZE);

    // 连接管理
    private final Map<SocketChannel, Long> lastActiveTime = new ConcurrentHashMap<>();
    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
    private final ExecutorService workerPool = new ThreadPoolExecutor(
        Runtime.getRuntime().availableProcessors(),
        Runtime.getRuntime().availableProcessors() * 2,
        60, TimeUnit.SECONDS,
        new SynchronousQueue<>(),
        new ThreadFactory() {
            private final AtomicInteger counter = new AtomicInteger(1);
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r, "server-worker-" + counter.getAndIncrement());
                thread.setDaemon(true);
                return thread;
            }
        }
    );

    public InterfaceServer() throws IOException {
        // 初始化NIO
        selector = Selector.open();
        serverChannel = ServerSocketChannel.open();
        serverChannel.configureBlocking(false);
        serverChannel.socket().bind(new InetSocketAddress(PORT));
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);

        System.out.println("接口服务器启动，监听端口: " + PORT);
        System.out.println("支持的接口:");
        System.out.println("- " + INTERFACE_USER + "?userId=xxx");
        System.out.println("- " + INTERFACE_ORDER + "?orderId=xxx");
        System.out.println("- " + INTERFACE_PRODUCT + "?productId=xxx");

        // 启动心跳检测
        scheduler.scheduleAtFixedRate(this::checkHeartbeatTimeout,
            HEARTBEAT_TIMEOUT, HEARTBEAT_TIMEOUT, TimeUnit.MILLISECONDS);
    }

    public void start() {
        try {
            while (!Thread.currentThread().isInterrupted()) {
                int readyChannels = selector.select(1000);
                if (readyChannels == 0) continue;

                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> keyIterator = selectedKeys.iterator();

                while (keyIterator.hasNext()) {
                    SelectionKey key = keyIterator.next();
                    keyIterator.remove();

                    try {
                        if (key.isAcceptable()) {
                            handleAccept(key);
                        } else if (key.isReadable()) {
                            handleRead(key);
                        }
                    } catch (IOException e) {
                        closeChannel(key.channel());
                    }
                }
            }
        } catch (IOException e) {
            System.err.println("服务器异常: " + e.getMessage());
        } finally {
            shutdown();
        }
    }

    // 处理新连接
    private void handleAccept(SelectionKey key) throws IOException {
        ServerSocketChannel server = (ServerSocketChannel) key.channel();
        SocketChannel clientChannel = server.accept();
        clientChannel.configureBlocking(false);
        clientChannel.register(selector, SelectionKey.OP_READ);

        // 记录连接信息和时间
        lastActiveTime.put(clientChannel, System.currentTimeMillis());
        SocketAddress clientAddr = clientChannel.getRemoteAddress();
        System.out.println("新客户端连接: " + clientAddr + 
            "，当前连接数: " + lastActiveTime.size());
    }

    // 处理客户端请求
    private void handleRead(SelectionKey key) throws IOException {
        SocketChannel clientChannel = (SocketChannel) key.channel();
        readBuffer.clear();

        int bytesRead = clientChannel.read(readBuffer);
        if (bytesRead == -1) {
            // 客户端主动断开
            closeChannel(clientChannel);
            return;
        }

        // 更新活跃时间
        lastActiveTime.put(clientChannel, System.currentTimeMillis());

        // 解析请求
        readBuffer.flip();
        String request = StandardCharsets.UTF_8.decode(readBuffer).toString().trim();
        
        // 处理心跳
        if ("HEARTBEAT".equals(request)) {
            return; // 心跳不需要回复
        }

        // 异步处理接口请求
        workerPool.submit(() -> processInterfaceRequest(clientChannel, request));
    }

    // 处理接口请求
    private void processInterfaceRequest(SocketChannel client, String request) {
        try {
            String response;
            
            // 路由到对应的接口处理
            if (request.startsWith(INTERFACE_USER)) {
                response = handleUserInterface(request);
            } else if (request.startsWith(INTERFACE_ORDER)) {
                response = handleOrderInterface(request);
            } else if (request.startsWith(INTERFACE_PRODUCT)) {
                response = handleProductInterface(request);
            } else {
                response = "错误: 不支持的接口 - " + request;
            }

            // 发送响应
            sendResponse(client, response);

        } catch (IOException e) {
            System.err.println("处理请求异常: " + e.getMessage());
            closeChannel(client);
        }
    }

    // 解析请求参数
    private Map<String, String> parseParams(String request) {
        Map<String, String> params = new HashMap<>();
        if (!request.contains("?")) {
            return params;
        }

        String[] parts = request.split("\\?", 2);
        if (parts.length < 2) {
            return params;
        }

        String[] paramPairs = parts[1].split("&");
        for (String pair : paramPairs) {
            String[] keyValue = pair.split("=", 2);
            if (keyValue.length == 2) {
                params.put(keyValue[0], keyValue[1]);
            }
        }
        return params;
    }

    // 处理用户接口
    private String handleUserInterface(String request) {
        Map<String, String> params = parseParams(request);
        String userId = params.get("userId");
        
        if (userId == null || userId.isEmpty()) {
            return "错误: 用户接口缺少userId参数";
        }
        
        // 模拟查询用户信息
        return String.format("用户信息 [接口: %s]: {userId: %s, name: '用户%s', status: 'active'}",
            INTERFACE_USER, userId, userId);
    }

    // 处理订单接口
    private String handleOrderInterface(String request) {
        Map<String, String> params = parseParams(request);
        String orderId = params.get("orderId");
        
        if (orderId == null || orderId.isEmpty()) {
            return "错误: 订单接口缺少orderId参数";
        }
        
        // 模拟查询订单信息
        return String.format("订单信息 [接口: %s]: {orderId: %s, amount: %.2f, date: '2023-10-01'}",
            INTERFACE_ORDER, orderId, 100 + (Integer.parseInt(orderId) % 900));
    }

    // 处理产品接口
    private String handleProductInterface(String request) {
        Map<String, String> params = parseParams(request);
        String productId = params.get("productId");
        
        if (productId == null || productId.isEmpty()) {
            return "错误: 产品接口缺少productId参数";
        }
        
        // 模拟查询产品信息
        return String.format("产品信息 [接口: %s]: {productId: %s, name: '产品%s', price: %.2f}",
            INTERFACE_PRODUCT, productId, productId, 50 + (Integer.parseInt(productId) % 150));
    }

    // 发送响应
    private void sendResponse(SocketChannel client, String response) throws IOException {
        if (!client.isOpen()) return;
        
        writeBuffer.clear();
        writeBuffer.put((response + "\n").getBytes(StandardCharsets.UTF_8));
        writeBuffer.flip();
        client.write(writeBuffer);
    }

    // 心跳超时检测
    private void checkHeartbeatTimeout() {
        long now = System.currentTimeMillis();
        List<SocketChannel> toRemove = new ArrayList<>();
        
        for (Map.Entry<SocketChannel, Long> entry : lastActiveTime.entrySet()) {
            if (now - entry.getValue() > HEARTBEAT_TIMEOUT) {
                toRemove.add(entry.getKey());
            }
        }
        
        // 关闭超时连接
        for (SocketChannel channel : toRemove) {
            closeChannel(channel);
        }
    }

    // 关闭通道
    private void closeChannel(Channel channel) {
        try {
            if (channel instanceof SocketChannel) {
                SocketChannel client = (SocketChannel) channel;
                System.out.println("客户端断开: " + client.getRemoteAddress() +
                    "，当前连接数: " + (lastActiveTime.size() - 1));
                lastActiveTime.remove(client);
            }
            channel.close();
        } catch (IOException e) {
            // 忽略关闭异常
        }
    }

    // 关闭服务器
    private void shutdown() {
        try {
            selector.close();
            serverChannel.close();
            scheduler.shutdown();
            workerPool.shutdown();
            System.out.println("服务器已关闭");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        try {
            new InterfaceServer().start();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
